package com.qingyan.flink.datasource.mysql;

import cn.hutool.core.bean.BeanUtil;
import com.qingyan.flink.core.properties.FlinkDataSourceProperties;
import com.qingyan.flink.core.sink.JdbcCatalogSinkBuilder;
import com.qingyan.flink.core.sink.SinkBuilder;

import java.util.Map;

/**
 * @author csp 2023-12-26
 */
public class MySQLSinkBuilder extends JdbcCatalogSinkBuilder {

    private MySqlDatasourceProperties datasourceProperties;

    public MySQLSinkBuilder() {
    }

    public MySQLSinkBuilder(FlinkDataSourceProperties sinkProperties, Map<String, String> sinkConfig) {
        super(sinkProperties, sinkConfig);
        datasourceProperties = new MySqlDatasourceProperties();
        BeanUtil.fillBeanWithMap(sinkProperties.getProperties(), datasourceProperties, true);
        catalog = new MySqlCatalogBuilder().builder()
                .username(datasourceProperties.getUsername())
                .pwd(datasourceProperties.getPassword())
                .host(datasourceProperties.getHostname())
                .port(datasourceProperties.getPort())
                .database(getSchema())
                .build();
    }
    @Override
    public String getHandle() {
        return "mysql";
    }



}
